package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.mapper.ApArticleConfigMapper;
import com.heima.common.constants.WmNewsMessageConstants;
import com.heima.model.article.pojos.ApArticleConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class ArtilceIsDownOrUpListener {

    @Autowired
    private ApArticleConfigMapper apArticleConfigMapper;

    @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
    public void downOrUp(ConsumerRecord<String,String> record){
        String value = record.value();
        if (StringUtils.isBlank(value)){
            return;
        }
        Map<String,String> map = JSON.parseObject(value, Map.class);
        long newsId = Long.parseLong(map.get("articleId"));
        short enable = Short.parseShort(map.get("enable"));

        ApArticleConfig apArticleConfig = apArticleConfigMapper.selectOne(Wrappers.lambdaQuery(ApArticleConfig.class).eq(ApArticleConfig::getArticleId, newsId));
        if (apArticleConfig==null){
            return;
        }
        apArticleConfig.setIsDown(enable==0?true:false);

        apArticleConfigMapper.updateById(apArticleConfig);
    }
}
